/*************************************************************************************
* 	 Copyright (C) 2010 by Information Systems Group, Saarland University  			*
*    http://infosys.cs.uni-saarland.de												*
* 	 																				*
* 	 This file is part of Hadoop++.												 	*
*																					*
*    Hadoop++ is free software: you can redistribute it and/or modify				*
*    it under the terms of the GNU Lesser General Public License as published by	*
*    the Free Software Foundation, either version 3 of the License, or				*
*    (at your option) any later version.											*
*																					*
*    Hadoop++ is distributed in the hope that it will be useful,					*
*    but WITHOUT ANY WARRANTY; without even the implied warranty of					*
*    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the					*
*    GNU Lesser General Public License for more details.							*
*																					*
*    You should have received a copy of the GNU Lesser General Public License		*
*    along with Hadoop++.  If not, see <http://www.gnu.org/licenses/>.				*
*************************************************************************************/
package tpc.benchmarks.joinOnCogroup;

import java.io.IOException;
import java.util.ArrayList;

import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RecordReader;

import unisb.cs.core.binary.reader.BinaryReaderInputFormat;
import unisb.cs.core.binary.utils.BinaryUtils;
import unisb.cs.core.join.cogroup.structure.GroupHeader;
import unisb.cs.data.tables.TableObjectProxy;

/**
 * This class set the uservisit.sourceIP and ranking.pageRank+UV.adRevenue 
 * as <key, value> pairs for the map function.
 */
public class JoinTaskRecordReader implements RecordReader<Text, Text> {
	
	// split variables
	protected FileSplit fileSplit;
	protected FSDataInputStream in;
	protected long splitSize = 0;
	protected long offset = 0;
	protected long start;
	protected long splitId;
	
	protected int sizeHeaderLen = 1;
	protected byte[] sizeHeader = new byte[sizeHeaderLen];
	protected String leftTableName;
	protected String rightTableName;
	protected ArrayList<String> leftRecordSet;
	protected ArrayList<String> rightRecordSet;
	protected byte[] leftData;
	protected byte[] rightData;
	protected int leftRecordSetOffset;
	protected int rightRecordSetOffset;
	protected byte[] keyBytes;

	public JoinTaskRecordReader(FileSplit fileSplit, JobConf job)
			throws IOException {
		
		this.leftTableName = job.get("left.table.name");
		this.rightTableName = job.get("right.table.name");
		
		this.fileSplit = fileSplit;
		Path file = fileSplit.getPath();
		FileSystem fs = file.getFileSystem(job);
		this.in = fs.open(file);
		splitSize = fileSplit.getLength();
		start = fileSplit.getStart();
		splitId = Integer.parseInt(job.get(BinaryReaderInputFormat.splitId(file.getName(), start)));

		IOUtils.skipFully(in, fileSplit.getStart());
		fetchNextGroup();
	}
	
	//һ�ζ��������������һ��group������ڴ�Ų��¾���Ҫ�޸ĳ�һ�ζ���һ����Ȼ��ÿ������һ��һ������ɼ�¼
	protected void fetchNextGroup() throws IOException {

		this.leftRecordSet = new ArrayList<String>();
		this.rightRecordSet = new ArrayList<String>();
		this.leftRecordSetOffset = 0;
		this.rightRecordSetOffset = 0;
		
		readData(leftData);
		getRecordSet(leftData, leftRecordSet, leftTableName);
		readData(rightData);
		getRecordSet(rightData, rightRecordSet, rightTableName);
	}
	
	private void readData(byte[] dataArray) throws IOException{
		
		IOUtils.readFully(in, sizeHeader, 0, sizeHeaderLen);
		offset += sizeHeaderLen;

		int headerSize = GroupHeader.getSize(sizeHeader);
		byte[] header = new byte[headerSize];
		IOUtils.readFully(in, header, 0, headerSize);
		offset += headerSize;

		GroupHeader gh = new GroupHeader(header);
		keyBytes = gh.getGroupValue();
		dataArray = new byte[gh.getGroupSize()];
		IOUtils.readFully(in, dataArray, 0, gh.getGroupSize());
		offset += gh.getGroupSize();
	}
	
	private void getRecordSet(byte[] dataArray, ArrayList<String> recordSet, String tableName){
		
		int groupOffset = 0;
		int groupLength = dataArray.length;
		
		while(groupOffset < groupLength){
			
			groupOffset = groupOffset + 1;  //�������ұ��־
			
			TableObjectProxy proxy = null;
			int recordLen = -1;
			
			try {
				proxy = new TableObjectProxy(tableName);
				recordLen = proxy.getSize();
			} catch (Exception e) {
				e.printStackTrace();
			}
			if (groupOffset > (groupLength - recordLen)) {
				return;
			}

			byte[] data = BinaryUtils.getBytes(dataArray, groupOffset, recordLen);
			groupOffset += recordLen;
			try {
				proxy.setRecord(data);
				if(doSelection() == true)
					recordSet.add(proxy.getStringRecord());
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}
	
	public boolean doSelection(){
		return true;	
	}

	@Override
	public Text createKey() {
		return new Text();
	}

	@Override
	public Text createValue() {
		return new Text();
	}

	@Override
	public long getPos() throws IOException {
		return offset;
	}

	@Override
	public float getProgress() throws IOException {
		return in.getPos() / (float) fileSplit.getLength();
	}

	@Override
	public boolean next(Text key, Text value) throws IOException {
		if(offset < splitSize){			
			if(this.leftRecordSetOffset < this.leftRecordSet.size() && 
					this.rightRecordSetOffset >= this.rightRecordSet.size()){
				this.leftRecordSetOffset++;
				this.rightRecordSetOffset = 0;
			}
			
			if(this.leftRecordSetOffset >= this.leftRecordSet.size() && 
					this.rightRecordSetOffset >= this.rightRecordSet.size()){
				fetchNextGroup();
			}
			
			byte[] k1Bytes = BinaryUtils.toBytes(splitId);
			key.append(k1Bytes, 0, 8);
			key.append(keyBytes, 0, keyBytes.length);
			value.set(this.leftRecordSet.get(leftRecordSetOffset) + this.rightRecordSet.get(rightRecordSetOffset));
			return true;
		}
		return false;
	}

	@Override
	public void close() throws IOException {
		IOUtils.closeStream(in);
	}
}
